package com.guchenbo.flink;

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * @author guchenbo
 * @date 2024/6/15
 */
public class TransactionDemo {
    public static void main(String[] args) {
        String inputTopic = "gcb-datagen";
        String outputTopic = "flink_output";
        String consumerGroup = "gcb-flink";
        String bootstrapServers = "kafka:9092";

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);

        // source
        KafkaSource<String> kafkaSource = KafkaSource.<String>builder().setBootstrapServers(bootstrapServers)
                        .setTopics(inputTopic).setGroupId(consumerGroup)
                        .setStartingOffsets(OffsetsInitializer.earliest())
                        .setValueOnlyDeserializer(new SimpleStringSchema()).build();
        DataStream<String> dataStream = env.fromSource(kafkaSource, WatermarkStrategy.noWatermarks(),
                        "Kafka Source");
        // 算子
        TransactionCodec transactionCodec = new TransactionCodec();
        DataStream<Transaction> transactionDataStream = dataStream.map(transactionCodec::fromCsv);

    }
}
